package com.ds.lens.flink.monitor.monitorjob.function;

import com.ds.lens.data.common.document.dynamic.DynamicDetectionResult;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.JacksonConfiguration;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-11
 */
@Slf4j
public class DynamicDetectionResultSinkFunction implements ElasticsearchSinkFunction<DynamicDetectionResult> {
    private static final long serialVersionUID = 9060064720838287598L;

    @Override
    public void process(DynamicDetectionResult detectionResult, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
        if (detectionResult == null) {
            return;
        }
        String indexName = MonitorJobContext.getInstance().getForecastConfig().getEsDynamicDetectionResultIndex();
        String type = MonitorJobContext.getInstance().getForecastConfig().getEsDynamicDetectionResultType();
        try {
            requestIndexer.add(Requests.indexRequest()
                    .index(indexName)
                    .type(type)
                    .source(JacksonConfiguration.newJsonObjectMapper().writeValueAsBytes(detectionResult), XContentType.JSON));
        } catch (JsonProcessingException e) {
            log.error("DynamicDetectionResultSinkFunction sink failed.", e);
        }
    }
}
